package com.zhouzj.task.zookeeper.consumer.handler;

import com.zhouzj.task.zookeeper.common.consist.Beat;
import com.zhouzj.task.zookeeper.common.metrics.RequestMetrics;
import com.zhouzj.task.zookeeper.common.rpc.RpcResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;

import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.SynchronousQueue;

/**
 * @Author: zhouzhijun
 * @Date: 2021/7/6 11:47
 */
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    private Map<String, SynchronousQueue<RpcResponse>> results;

    public RpcClientHandler(Map<String, SynchronousQueue<RpcResponse>> results) {
        this.results = results;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcResponse rpcResponse) throws Exception {
        RequestMetrics.getInstance().calculate(rpcResponse.getRequestId());
        System.out.println("请求id:" + rpcResponse.getRequestId() + ", 返回结果:" + rpcResponse.getResult());
        if(!rpcResponse.getSuccess()){
            throw new RuntimeException("调用结果异常,异常信息:" + rpcResponse.getError());
        }

        if(!rpcResponse.getSuccess()){
            throw new RuntimeException("调用结果异常,异常信息:" + rpcResponse.getError());
        }
        // 取出结果容器,将response放进queue中
        SynchronousQueue<RpcResponse> queue = results.get(rpcResponse.getRequestId());
        queue.put(rpcResponse);
    }

    /**
     * 定时处理
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.writeAndFlush(Beat.BEAT_PING);
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
